package sbc.ms;

import java.util.List;

import org.xvsm.api.core.Capi;
import org.xvsm.api.core.ContainerRef;
import org.xvsm.api.core.CoordinationTypes;
import org.xvsm.api.core.Entry;
import org.xvsm.api.core.Tuple;
import org.xvsm.client.ClientCapi;
import org.xvsm.ms.exceptions.FatalException;

public class Taker extends Thread {
	
	public static void main (String [] args)  {
		if (args.length != 1) {
			System.out.println("URL needed!");
			System.out.println("Usage: java -cp <classpath> sbc.ms.Taker http://localhost:8080/xvsm/xvsm");
			System.exit(1);
		}
		
		/*
		 * get Mozart Space Container
		 */
		Capi capi = new ClientCapi(args[0]);
		capi.init(false);
		ContainerRef cref;
		try {
			cref = Util.getOrCreateNamedContainer(capi, "Container", CoordinationTypes.FIFO);
			
	        /*
			 * take every tupel out of Mozart Space
			 */
			Taker taker = new Taker(capi, cref);
			taker.start();
		}
		catch (FatalException e) {
			e.printStackTrace();
		}
	}
	
    private Capi capi;
    private ContainerRef cref;
    
	public Taker (Capi capi, ContainerRef cref) {
		System.out.println ("Taker init");
		this.capi = capi;
		this.cref = cref;
	}
	
	@Override
	public void run() {
		System.out.println("Taker up and running");
		
		/*
		 * take all entries out of Mozart Space
		 * (including deletion)
		 */
		try {
			for (;;) {
				System.out.println("take all entries out of Mozart Space");
				List<Entry> take = this.capi.take(this.cref);
				
				Tuple t = (Tuple) take.get(0);
				
				List<Entry> elements = t.getValue();
		
				System.out.println(elements.get(0).getValue() + " "
					+ elements.get(1).getValue());
			}
		}
		catch (Exception e)	{
			System.out.println("e: " + e);
		}		
	}
}
